package day09

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Milliseconds, Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}


object windowDemo {
  /**
    * 把一个批次间隔为2秒的  ，使用窗口 展示 每10秒计算一次，一次计算10秒的
    *
    */
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("zzz").setMaster("local[2]")
    val ssc = new StreamingContext(conf, Milliseconds(2000))
    val dStream: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)
    val tups: DStream[(String, Int)] = dStream.flatMap(_.split(" ")).map((_,1))

    // 使用窗口reduceByKeyAndWindow(函数,窗口大小，窗口步长即滑动间隔)，注意window第一个参数为函数，
    // window的方法并不能推断出该函数的参数类型。因此必须要指定该函数的参数类型。
    val res: DStream[(String, Int)] = tups.reduceByKeyAndWindow((x:Int, y:Int) => x+y, Seconds(10), Seconds(10))

    res.print()
    ssc.start()
    ssc.awaitTermination()

  }
}
